package com.lvy.hczn.mq.rocket;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 实现事务的监听接口
 *
 * @author yw
 * @date 2021-06-11 11:07:12
 **/
@Slf4j
public class TransactionListenerImpl implements TransactionListener {

    /**
     * 事务消息使用上的限制:
     * 1.事务消息不支持延时消息和批量消息。
     * 2.为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionalMessageCheckListener 类来修改这个行为。
     * 3.事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionTimeout 参数。
     * 4.事务性消息可能不止一次被检查或消费。
     * 5.提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
     * 6.事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。
     **/
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    /**
     * 当发送半消息成功时，使用 executeLocalTransaction 方法来执行本地事务
     *
     * @param message
     * @param o
     * @return
     * @author yw
     * @date 2021-06-11 11:48:43
     **/
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
        //测试，随机生成数字
        int value = transactionIndex.getAndIncrement();
        //status:0/1/2
        int status = value % 3;
        //TODO 业务逻辑处理，返回UNKNOW还是COMMIT_MESSAGE还是ROLLBACK_MESSAGE
        //TODO 可以根据message中的topic和tag，if else判断是什么数据走什么样的处理
        //TODO 如有多个需要分布式事物存储的，就用一个topic,多个tag区分
        //localTrans用redis代替
        localTrans.put(message.getTransactionId(), status);
        if (status == 1) {
            localTransactionState = LocalTransactionState.COMMIT_MESSAGE;
        } else if (status == 2) {
            localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        log.info("执行本地事务>>>value=" + value + ";status=" + status);
        log.info("执行本地事务>>>>localTrans=" + JSONUtil.toJsonStr(localTrans));
        log.info("执行本地事务>>>>msg=" + JSONUtil.toJsonStr(message));
        return localTransactionState;
    }

    /**
     * checkLocalTransaction 方法用于检查本地事务状态，并回应消息队列的检查请求
     * 单个消息的检查次数限制为 15 次
     *
     * @param msg
     * @return
     * @author yw
     * @date 2021-06-11 11:49:50
     **/
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        //redis获取
        Integer status = localTrans.get(msg.getTransactionId());
        log.info("检查本地事务状态>>>status=" + status);
        log.info("检查本地事务状态>>>msg=" + JSONUtil.toJsonStr(msg));
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
